
#include "spserver.h"
#include "Acceptor.h"
#include "Session.h"
#include "NetworkObject.h"
#include "SessionList.h"
#include "SessionPool.h"
#include "IoHandler.h"

void IoRoutine(void* handler)
{
	IoHandler* pIoHandler = (IoHandler*)handler;
	int kq = pIoHandler->GetIoSystemHandler();

	struct kevent events[MAX_EVENT_COUNT];
	while(true) 
	{
		int ret = kevent(kq, NULL, 0, events, MAX_EVENT_COUNT, NULL);

		for (int i = 0; i < ret; ++i)
		{

			{
				Session* pSession = (Session*)events[i].udata;
				if ((events[i].flags & EV_ERROR) || (events[i].flags & EV_EOF))
				{
					printf("socket %d error\n", pSession->GetSocket());
					pSession->Disconnect(true);

				}
				else
				{
					if (events[i].flags & EVFILT_READ)
					{
						//printf("socket %d recved data\n", pSession->GetSocket());
						unsigned char* pBytes = NULL;
						size_t nLen = 0;
						pSession->GetRecvBuffer()->GetRecvParam(&pBytes, nLen);
						nLen = min(events[i].data, nLen);
						if (nLen)
						{
							recv(pSession->GetSocket(), pBytes, nLen, 0);
							pSession->GetRecvBuffer()->Completion(nLen);
						}
					}
					if (events[i].flags & EVFILT_WRITE)
					{
						//printf("socket %d sended data\n", pSession->GetSocket());
						unsigned char* pBytes = NULL;
						size_t nLen = 0;
						pSession->GetSendBuffer()->GetSendParam(&pBytes, nLen);
						if (nLen)
						{
							send(pSession->GetSocket(), pBytes, nLen, 0);
							pSession->GetSendBuffer()->Completion(nLen);
						}
					}

				}
			}
		}

	}

}

void SendRoutine(void* handler)
{

}

IoHandler::IoHandler()
{

}

bool IoHandler::Init(size_t nIoThreads, size_t nMaxAcceptSession, size_t nMaxPacket, 
					 fnCallbackCreateAcceptedObject cbCreateObject, fnCallbackDestoryAcceptedObject cbDestoryObject
{
	cbCreate = cbCreateObject;
	cbDestory = cbDestoryObject;
	nMaxPacketSize = nMaxPacket;

	kq = kqueue();
	pAcceptor = new Acceptor();
	pAcceptor->init(this);
	pAcceptSessionPool = SessionPool::create(nMaxAcceptSession, 256*1024, 1024*1024, nMaxPacket);

	pAcceptedSessionList = new SessionList();
	pActiveSessionList = new SessionList();
	pTmpSessionList = new SessionList();


	for (size_t i = 0; i < nIoThreads; ++i)
	{
		ioThreads[i].start(IoRoutine, this);		
	}

	sendThread.start(SendRoutine, this);

	return true;
}

bool IoHandler::startListen(const char* szIp, int port)
{
	return pAcceptor->startListen(szIp, port);
}


Session* IoHandler::AllocAcceptSession()
{
	return pAcceptSessionPool->Alloc();
}


bool IoHandler::RegisterSession(Session* pSession)
{

	struct kevent changes[1];
	EV_SET(&changes[0], pSession->GetSocket(), EVFILT_READ, EV_ADD, 0, 0, pSession);
	int ret = kevent(kq, changes, 1, NULL, 0, NULL);
	return true;
}

bool IoHandler::ReuseSession(Session* pSession)
{

	close(pSession->GetSocket());
	pSession->SetSocket(retired_fd);
	pAcceptSessionPool->Free(pSession);

	return true;
}

void IoHandler::Update()
{
	ProcessActiveSessionList();

	if (pAcceptedSessionList->empty() == false)
		ProcessAcceptedSessionList();

	KickDeadSessions();
}

void IoHandler::ProcessAcceptedSessionList()
{
	pAcceptedSessionList->lock();
	pTmpSessionList->splice(pTmpSessionList->end(), *pAcceptedSessionList);
	pAcceptedSessionList->unlock();

	for (SessionList::iterator iter = pTmpSessionList->begin(); iter != pTmpSessionList->end();)
	{
		Session* pSession = *iter;
		SessionList::iterator iterTmp = iter;
		++iter;

		if (!RegisterSession(pSession))
		{
			pTmpSessionList->erase(iterTmp);
			ReuseSession(pSession);
			continue;
		}

		NetworkObject* pNetworkObject = cbCreate();
		if (!pNetworkObject)
		{
			pTmpSessionList->erase(iterTmp);
			ReuseSession(pSession);
			continue;
		}

		pSession->BindNetworkObject(pNetworkObject);
		pSession->OnAccept();
	}

	if (pTmpSessionList->empty() == false)
	{
		pActiveSessionList->lock();
		pActiveSessionList->splice(pActiveSessionList->end(), *pTmpSessionList);
		pActiveSessionList->unlock();
	}
}

void IoHandler::ProcessActiveSessionList()
{
	for (SessionList::iterator iter = pActiveSessionList->begin(); iter != pActiveSessionList->end(); ++iter)
	{
		Session* pSession = *iter;

		if (pSession->ShouleBeRemove())
			continue;

		if (!pSession->ProcessRecvdPacket(nMaxPacketSize))
		{
			pSession->Disconnect(true);
		}

	}

}

void IoHandler::KickDeadSessions()
{

	pActiveSessionList->lock();

	for (SessionList::iterator iter = pActiveSessionList->begin(); iter != pActiveSessionList->end();)
	{
		Session* pSession = *iter;
		SessionList::iterator iterTmp = iter;
		++iter;

		if (pSession->ShouleBeRemove())
		{
			pActiveSessionList->erase(iterTmp);
			pTmpSessionList->push_back(pSession);
		}
	}
	pActiveSessionList->unlock();
	
	for (SessionList::iterator iter = pTmpSessionList->begin(); iter != pTmpSessionList->end(); ++iter)
	{
		Session* pSession = *iter;
		
		NetworkObject* pNetwork = pSession->GetNetworkObject();
		pSession->UnBindNetworkObject();

		ReuseSession(pSession);

		if (pNetwork)
		{
			pNetwork->OnDisconnect();
			cbDestory(pNetwork);
		}
	}
	pTmpSessionList->clear();
}
